package com.ml.ws.service.impl;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSON;
import com.ml.enums.RequestType;
import com.ml.util.KlineDataUtils;
import com.ml.ws.model.PeriodLastKline;
import com.ml.ws.model.RequestParam;
import com.ml.ws.model.ResponseMessage;
import com.ml.ws.service.RequestFactory;
import com.ml.ws.service.RequestService;
import com.virgil.core.utils.GZipUtils;
import com.virgil.redis.util.RedisUtil;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;

@Service
public class ReqKline implements RequestService, InitializingBean {
	
	@Autowired
	RedisUtil redisUtil;
	
    @Override
    public void pushReqData(Channel channel, RequestParam param) {
        // 起始时间 from 默认值：当前时间前30分钟
        Long from = param.getFrom();
        // 截止时间 to 默认值：当前时间
        Long to = param.getTo();
        if (from == null || from == 0L) {
            from = System.currentTimeMillis() / 1000 - 60 * 60 * 24 * 7;
        }
        if (to == null || to == 0L) {
            to = System.currentTimeMillis() / 1000;
        }
        Map<String, Object> map = history(param.getSymbol(), param.getPeriod(), to, from);
        try {
            // 封装消息体
            ResponseMessage message = new ResponseMessage("kline_history", map);
            // 获取压缩后的数据
            byte[] bytes = GZipUtils.compress(JSON.toJSONString(message));
            // 转换成二进制数据
            ByteBuf buf = Unpooled.wrappedBuffer(bytes);
            channel.writeAndFlush(new BinaryWebSocketFrame(buf));
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        RequestFactory.register(RequestType.KLINE, this);
    }

    /**
     * k线历史数据
     *
     * @return
     */
    private Map<String, Object> history(String symbol, String periods, Long to, Long from) {
        Map<String, Object> map = new HashMap<>();
        String klineKey = StringUtils.join(symbol, ":klineData:back:", KlineDataUtils.getPeriod(periods));
        Long count = redisUtil.zcount(klineKey, from.intValue(), to.intValue());

        if (count.intValue() <= 0) {
            // TODO 开线程从数据库同步到redis 预留

        } else {
            Set<String> history = redisUtil.zrangeByScore(klineKey, to, from);
            List<PeriodLastKline> klines = new ArrayList<PeriodLastKline>();
            for (String string : history) {
                klines.add(JSON.parseObject(string, PeriodLastKline.class));
            }
            // 根据k线的时间节点排序
            klines.sort(new Comparator<PeriodLastKline>() {
                @Override
                public int compare(PeriodLastKline o1, PeriodLastKline o2) {
                    if (o1.getStartTime() > o2.getStartTime()) {
                        return -1;
                    }
                    return 1;
                }
            });
            if (klines.size() > 500) {
                klines = klines.subList(0, 500);
            }
            map.put("data", klines);
        }
        return map;
    }


}
